package net.bwie.realtime.jtp.log.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/21 09:52:51
 * @version: 1.0
 */
public class JtpTrafficAppStartMinuteWindowDwsJob {
    public static void main(String[] args) {
        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv();

        // 2. 输入表-input：映射到Kafka消息队列
        createInputTable(tabEnv);

        // 3. 数据处理-select
        Table reportTable2 = handle(tabEnv);

        // 4. 输出表-output：映射到Clickhouse表
        createOutputTable(tabEnv);


        // 5. 保存数据-insert
        saveToClickHouse(tabEnv, reportTable2) ;

        joinDim(tabEnv);
    }


    private static void joinDim(TableEnvironment tabEnv) {
        Table startAdResultTable = tabEnv.sqlQuery(
                "SELECT " +
                        "  window_start as window_start_time\n" +
                        ", window_end as window_end_time\n" +
                        ", open_ad_id\n" +
                        ", count(open_ad_id) as open_ad_count\n" +
                        ", sum(open_ad_ms) as open_ad_ms\n" +
                        ", sum(open_ad_skip_ms) as open_ad_skip_ms\n" +
                        ", UNIX_TIMESTAMP() * 1000 AS ts\n" +
                        ", PROCTIME() AS proc_time\n" +
                        "  FROM TABLE (\n" +
                        " TUMBLE(TABLE start_log_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES)\n" +
                        ") t1 GROUP BY window_start, window_end, open_ad_id"
        );

        tabEnv.createTemporaryView("start_ad_result_table", startAdResultTable);

        tabEnv.executeSql("CREATE TABLE `ads_info_mysql_source_dim`\n" +
                "(\n" +
                "    id               INT COMMENT '主键ID',\n" +
                "    name             STRING COMMENT '广告名称',\n" +
                "    content          STRING COMMENT '广告内容（图片URL/视频URL/H5代码）',\n" +
                "    type             STRING COMMENT '广告类型（IMAGE, VIDEO, HTML5）',\n" +
                "    start_time       TIMESTAMP COMMENT '展示开始时间',\n" +
                "    end_time         TIMESTAMP COMMENT '展示结束时间',\n" +
                "    is_active        INT COMMENT '是否启用',\n" +
                "    target_condition STRING COMMENT '定向条件（JSON格式）',\n" +
                "    created_at       TIMESTAMP COMMENT '数据插入时间',\n" +
                "    updated_at       TIMESTAMP COMMENT '数据更新时间',\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "      'connector' = 'jdbc',\n" +
                "      'url' = 'jdbc:mysql://node101:3306/jtp_ads',\n" +
                "      'table-name' = 'ads_info',\n" +
                "      'lookup.cache.max-rows' = '10',\n" +
                "      'lookup.cache.ttl' = '1 hour',\n" +
                "      'username' = 'root',\n" +
                "      'password' = '123456',\n" +
                "      'driver' = 'com.mysql.cj.jdbc.Driver'\n" +
                ")");

        // 确保连接条件和数据类型正确
        Table joinTable = tabEnv.sqlQuery("SELECT\n" +
                "    t1.window_start_time,\n" +
                "    t1.window_end_time,\n" +
                "    t2.name AS open_ad_name,\n" +
                "    t1.open_ad_count,\n" +
                "    t1.open_ad_ms,\n" +
                "    t1.open_ad_skip_ms,\n" +
                "    t1.ts\n" +
                "FROM start_ad_result_table t1\n" +
                "LEFT JOIN ads_info_mysql_source_dim FOR SYSTEM_TIME AS OF t1.proc_time AS t2\n" +
                "ON cast(t1.open_ad_id as INTEGER)  = t2.id");
        joinTable.execute().print();
    }

    private static void saveToClickHouse(TableEnvironment tabEnv, Table reportTable) {
        // 2-保存：APP启动类型统计

        tabEnv.createTemporaryView("report_table", reportTable);


//        tabEnv.executeSql(
//                "INSERT INTO dws_traffic_app_start_window_report_clickhouse_sink\n" +
//                        "SELECT\n" +
//                        "    DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time\n" +
//                        "    , DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time\n" +
//                        "    , entry\n" +
//                        "    , entry_count\n" +
//                        "    , loading_time\n" +
//                        "    , ts\n" +
//                        "FROM report_table\n"
//        );


        // 2-保存：APP启动广告统计
        tabEnv.executeSql(
                "INSERT INTO dws_traffic_app_start_ad_window_report_clickhouse_sink\n" +
                        "SELECT\n" +
                        "    DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time\n" +
                        "     , DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time\n" +
                        "     , open_ad_id\n" +
                        "     , open_ad_count\n" +
                        "     , open_ad_ms\n" +
                        "     , open_ad_skip_ms\n" +
                        "     , ts\n" +
                        "FROM report_table"
        );
    }

    private static void createOutputTable(TableEnvironment tabEnv) {
        // 1-app启动类型汇总表：映射ClickHouse

//        tabEnv.executeSql(
//        "CREATE TABLE dws_traffic_app_start_window_report_clickhouse_sink (\n" +
//                "  `window_start_time` STRING COMMENT '窗口开始日期时间',\n" +
//                "  `window_end_time` STRING COMMENT '窗口结束日期时间',\n" +
//                "  `entry` STRING COMMENT '启动app方式',\n" +
//                "  `entry_count` BIGINT COMMENT '启动app方式次数',\n" +
//                "  `loading_time` BIGINT COMMENT '启动app方式次数',\n" +
//                "  `ts` BIGINT COMMENT '启动app加载时间'\n" +
//                ") WITH (\n" +
//                "  'connector' = 'clickhouse',\n" +
//                "  'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',\n" +
//                "  'table' = 'dws_traffic_app_start_window_report',\n" +
//                "  'username' = 'default',\n" +
//                "  'password' = '',\n" +
//                "  'format' = 'json'\n" +
//                ")"
//        );

        // 2-app启动广告汇总表：映射ClickHouse
        tabEnv.executeSql(
                "CREATE TABLE dws_traffic_app_start_ad_window_report_clickhouse_sink (\n" +
                        "  `window_start_time` STRING COMMENT '窗口开始日期时间',\n" +
                        "  `window_end_time` STRING COMMENT '窗口结束日期时间',\n" +
                        "  `open_ad_id` STRING COMMENT '启动app方式',\n" +
                        "  `open_ad_count` BIGINT COMMENT '广告展示次数',\n" +
                        "  `open_ad_ms` BIGINT COMMENT '广告总共播放时间',\n" +
                        "  `open_ad_skip_ms` BIGINT COMMENT '用户跳过广告时间',\n" +
                        "  `ts` BIGINT COMMENT '数据计算时间戳'\n" +
                        ") WITH (\n" +
                        "  'connector' = 'clickhouse',\n" +
                        "  'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',\n" +
                        "  'table' = 'dws_traffic_app_start_ad_window_report',\n" +
                        "  'username' = 'default',\n" +
                        "  'password' = '',\n" +
                        "  'format' = 'json'\n" +
                        ")"
        );
    }


    private static Table handle(TableEnvironment tabEnv) {

//       todo:s2-APP启动类型统计：设置窗口进行分组、聚合计算
        {  // s1-提取启动日志字段值
            Table startLogTable = tabEnv.sqlQuery(
                    "SELECT\n" +
                            " `start`['entry'] AS entry\n" +
                            ", CAST(`start`['loading_time'] AS BIGINT) AS loading_time\n" +
                            ", `start`['open_ad_id'] AS open_ad_id\n" +
                            ", CAST(`start`['open_ad_ms'] AS BIGINT) AS open_ad_ms\n" +
                            ", CAST(`start`['open_ad_skip_ms'] AS BIGINT) AS open_ad_skip_ms\n" +
                            ", row_time\n" +
                            "FROM dwd_traffic_start_log_kafka_source\n" +
                            "WHERE `start`['entry'] IS NOT NULL\n"
            );
            //startLogTable.execute().print();
            tabEnv.createTemporaryView("start_log_table", startLogTable);

            Table reportTable1 = tabEnv.sqlQuery(
                    "SELECT " +
                            "  TUMBLE_START(row_time, INTERVAL '1' MINUTES) AS window_start_time, " +
                            "  TUMBLE_END(row_time, INTERVAL '1' MINUTES) AS window_end_time, " +
                            "  entry," +
                            "  count(entry) as entry_count," +
                            "  loading_time," +
                            "  UNIX_TIMESTAMP() * 1000 AS ts" +
                            "  FROM start_log_table " +
                            "  GROUP BY TUMBLE(row_time, INTERVAL '1' MINUTES),entry,loading_time"
            );
        }

//       todo:s3-APP启动广告统计：设置窗口进行分组、聚合计算
        {
            Table startReportTable = tabEnv.sqlQuery(
                    "SELECT\n" +
                            " `start`['open_ad_id'] AS open_ad_id\n" +
                            ", CAST(`start`['open_ad_ms'] AS BIGINT) AS open_ad_ms\n" +
                            ", CAST(`start`['open_ad_skip_ms'] AS BIGINT) AS open_ad_skip_ms\n" +
                            ", row_time\n" +
                            " FROM dwd_traffic_start_log_kafka_source"
            );

            tabEnv.createTemporaryView("start_report_table", startReportTable);

            Table reportTable2 = tabEnv.sqlQuery(
                    "SELECT\n" +
                            "       TUMBLE_START(row_time, INTERVAL '1' MINUTES) AS window_start_time\n" +
                            "     , TUMBLE_END(row_time, INTERVAL '1' MINUTES) AS window_end_time\n" +
                            "     , open_ad_id\n" +
                            "     , count(open_ad_id) AS open_ad_count\n" +
                            "     , sum(open_ad_ms) AS open_ad_ms\n" +
                            "     , sum(open_ad_skip_ms) AS open_ad_skip_ms\n" +
                            "     , UNIX_TIMESTAMP() * 1000 AS ts\n" +
                            "       FROM start_log_table\n" +
                            "       GROUP BY\n" +
                            "       TUMBLE(row_time, INTERVAL '1' MINUTES ),\n" +
                            "       open_ad_id"
            );
//            reportTable2.execute().print();
            return reportTable2;
        }
    }

    private static void createInputTable(TableEnvironment tabEnv) {

//        输入表 通过 连接kfk  获取数据
        tabEnv.executeSql(
                "CREATE TABLE dwd_traffic_start_log_kafka_source (\n" +
                        "  `common` MAP<STRING, STRING> COMMENT '公共环境信息',\n" +
                        "  `start` MAP<STRING, STRING> COMMENT '启动信息',\n" +
                        "  `ts` BIGINT,\n" +
                        "  row_time AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000, 'yyyy-MM-dd HH:mm:ss.SSS')),\n" +
                        "  WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'dwd_traffic-start-log',\n" +
                        "  'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                        "  'properties.group.id' = 'gid_dws_log_app_start',\n" +
                        "  'scan.startup.mode' = 'earliest-offset',\n" +
                        "  'format' = 'json',\n" +
                        "  'json.fail-on-missing-field' = 'false',\n" +
                        "  'json.ignore-parse-errors' = 'true'\n" +
                        ")"
        );
        //tabEnv.executeSql("SELECT * FROM dwd_log_start_log_kafka_source LIMIT 10").print();

    }

    private static TableEnvironment getTableEnv() {
        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // 3返回对象
        return tabEnv;
    }


}
